package org.stream.answerduel.service;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.util.Random;

import javax.annotation.PostConstruct;


/**
 * 功能描述：
 *
 * @Auther: 苏建誉
 * @Date: 2019/7/09 22:12
 */
@Service
@Slf4j
public class MQService{

    /**
     * 消息质量
     */
    @Value("${mqtt.qos}")
    private int qos;

    /**
     * MQ服务器地址
     */
    @Value("${mqtt.url}")
    private String broker = null;

    /**
     * 客户ID
     */
    @Value("${mqtt.clientId}")
    private String clientId = null;

    /**
     * MQ客户端
     */
    private MqttClient client;

    /**
     * 内存存储
     */
    private MemoryPersistence persistence = new MemoryPersistence();

    private Random random = new Random();

    /**
     * 功能描述：初始化函数，负责连接MQ服务器
     *
     *
     */
    @PostConstruct
    public void MQInit(){
        try{
            // 创建客户端
            client = new MqttClient(broker, clientId+random.nextInt(1000), persistence);
            // 创建链接参数
            MqttConnectOptions connOpts = new MqttConnectOptions();
            // 在重新启动和重新连接时记住状态
            connOpts.setCleanSession(false);
            // 设置超时时间 单位为秒
            connOpts.setConnectionTimeout(60);
            // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
            connOpts.setKeepAliveInterval(20);
            // 建立连接
            client.connect(connOpts);
            subscribe("/broker/disconnection");
        }catch (MqttException me) {
            log.error("reason " + me.getReasonCode());
            log.error("msg " + me.getMessage());
            log.error("loc " + me.getLocalizedMessage());
            log.error("cause " + me.getCause());
            log.error("exception" + me);
            me.printStackTrace();
        }
    }

    /**
     * 功能描述：根据内容和主题向MQ服务器发布消息
     *
     * @param content 发布消息的内容
     * @param topic 发布主题
     */
    public void publish(String content,String topic) throws MqttException{
        if(!client.isConnected()){
            log.info("尝试连接MQ服务器中0.0");
            MQInit();
        }
        MqttMessage message = new MqttMessage(content.getBytes());
        // 设置消息的服务质量
        message.setQos(qos);
        // 发布消息
        client.publish(topic, message);
    }
    /**
     * 功能描述：订阅相关主题
     *
     * @param topic 订阅主题
     */
    public void subscribe(String topic) throws MqttException{
        //订阅消息
        client.subscribe(topic, qos);
        // 设置回调函数
        client.setCallback(new MqttCallback() {

            public void connectionLost(Throwable cause) {
                log.info("尝试连接MQ服务器中0.0");
                MQInit();
            }

            public void messageArrived(String topic, MqttMessage message){
                log.info("topic:"+topic);
                log.info("Qos:"+message.getQos());
                log.info("message content:"+new String(message.getPayload()));

            }

            public void deliveryComplete(IMqttDeliveryToken token) {
                //System.out.println("deliveryComplete---------"+ token.isComplete());
            }

        });
    }
}
